package com.ultrapower.util;

import java.util.ArrayList;
import java.util.List;

import com.ultrapower.bean.AnnotationBean;
import com.ultrapower.bean.BasicSpan;
import com.ultrapower.bean.SpanBean;
import com.ultrapower.bean.SpanChunkBean;
import com.ultrapower.bean.SpanEventBean;
import com.ultrapower.buffer.Buffer;
import com.ultrapower.util.span.AnnotationTranscoder;
import com.ultrapower.util.span.SpanBitFiled;
import com.ultrapower.util.span.SpanDecodingContext;
import com.ultrapower.util.span.SpanEventBitField;
import com.ultrapower.util.span.SpanEventQualifierBitField;

public class SpanDecoder {
	byte TYPE_SPAN = 0;
	byte TYPE_SPAN_CHUNK = 1;

	// reserved
	byte TYPE_PASSIVE_SPAN = 4;
	byte TYPE_INDEX = 7;

	private static final AnnotationTranscoder transcoder = new AnnotationTranscoder();

	public Object decode(Buffer qualifier, Buffer columnValue, SpanDecodingContext decodingContext) {
		final byte type = qualifier.readByte();

		if (TYPE_SPAN == type) {

			SpanBean span = readSpan(qualifier, columnValue, decodingContext);
			return span;

		} else if (TYPE_SPAN_CHUNK == type) {

			SpanChunkBean spanChunk = readSpanChunk(qualifier, columnValue, decodingContext);
			return spanChunk;

		} else {
			System.out.println("Unknown span type " + type);
			return "UNKNOWN";
		}
	}

	private SpanBean readSpan(Buffer qualifier, Buffer columnValue, SpanDecodingContext decodingContext) {
		final SpanBean span = new SpanBean();

		final TransactionId transactionId = decodingContext.getTransactionId();
		span.setTransactionId(transactionId);
		span.setCollectorAcceptTime(decodingContext.getCollectorAcceptedTime());

		SpanEventBean firstSpanEvent = readQualifier(span, qualifier);

		readSpanValue(columnValue, span, firstSpanEvent, decodingContext);

		return span;
	}

	private SpanEventBean readQualifier(BasicSpan basicSpan, Buffer buffer) {
		String applicationId = buffer.readPrefixedString();
		basicSpan.setApplicationId(applicationId);

		String agentId = buffer.readPrefixedString();
		basicSpan.setAgentId(agentId);

		long agentStartTime = buffer.readVLong();
		basicSpan.setAgentStartTime(agentStartTime);

		long spanId = buffer.readLong();
		basicSpan.setSpanId(spanId);

		int firstSpanEventSequence = buffer.readSVInt();
		if (firstSpanEventSequence == -1) {
			// buffer.readByte();
			// spanEvent not exist ??
			System.out.println("firstSpanEvent is null. bug!!!! firstSpanEventSequence:" + firstSpanEventSequence);
			throw new IllegalStateException("firstSpanEvent is null");
		} else {
			return readQualifierFirstSpanEvent(buffer);
		}
	}

	private SpanEventBean readQualifierFirstSpanEvent(Buffer buffer) {
		final SpanEventBean firstSpanEvent = new SpanEventBean();

		final byte bitField = buffer.readByte();
		if (SpanEventQualifierBitField.isSetAsync(bitField)) {
			int asyncId = buffer.readInt();
			int asyncSequence = buffer.readVInt();
			firstSpanEvent.setAsyncId(asyncId);
			firstSpanEvent.setAsyncSequence((short) asyncSequence);
		}
		return firstSpanEvent;
	}

	public void readSpanValue(Buffer buffer, SpanBean span, SpanEventBean firstSpanEvent,
			SpanDecodingContext decodingContext) {

		final byte version = buffer.readByte();
		if (version != 0) {
			throw new IllegalStateException("unknown version :" + version);
		}
		span.setVersion(version);

		final SpanBitFiled bitFiled = new SpanBitFiled(buffer.readByte());

		final short serviceType = buffer.readShort();
		span.setServiceType(serviceType);

		switch (bitFiled.getApplicationServiceTypeEncodingStrategy()) {
		case PREV_EQUALS:
			span.setApplicationServiceType(serviceType);
			break;
		case RAW:
			span.setApplicationServiceType(buffer.readShort());
			break;
		default:
			throw new IllegalStateException("applicationServiceType");
		}

		if (!bitFiled.isRoot()) {
			span.setParentSpanId(buffer.readLong());
		} else {
			span.setParentSpanId(-1);
		}

		final long startTimeDelta = buffer.readVLong();
		final long startTime = span.getCollectorAcceptTime() - startTimeDelta;
		span.setStartTime(startTime);
		span.setElapsed(buffer.readVInt());

		span.setRpc(buffer.readPrefixedString());

		span.setEndPoint(buffer.readPrefixedString());
		span.setRemoteAddr(buffer.readPrefixedString());
		span.setApiId(buffer.readSVInt());

		if (bitFiled.isSetErrorCode()) {
			span.setErrCode(buffer.readInt());
		}
		if (bitFiled.isSetHasException()) {
			int exceptionId = buffer.readSVInt();
			String exceptionMessage = buffer.readPrefixedString();
			span.setExceptionInfo(exceptionId, exceptionMessage);
		}

		if (bitFiled.isSetFlag()) {
			span.setFlag(buffer.readShort());
		}

		if (bitFiled.isSetLoggingTransactionInfo()) {
			span.setLoggingTransactionInfo(buffer.readByte());
		}

		span.setAcceptorHost(buffer.readPrefixedString());

		if (bitFiled.isSetAnnotation()) {
			List<AnnotationBean> annotationBoList = readAnnotationList(buffer, decodingContext);
			span.setAnnotationBoList(annotationBoList);
		}

		List<SpanEventBean> spanEventBoList = readSpanEvent(buffer, firstSpanEvent, decodingContext);
		span.addSpanEventBoList(spanEventBoList);

	}

	private List<AnnotationBean> readAnnotationList(Buffer buffer, SpanDecodingContext decodingContext) {
		int annotationListSize = buffer.readVInt();
		List<AnnotationBean> annotationBoList = new ArrayList<AnnotationBean>(annotationListSize);

		// AnnotationBo prev = decodingContext.getPrevFirstAnnotationBo();
		AnnotationBean prev = null;
		for (int i = 0; i < annotationListSize; i++) {
			AnnotationBean current;
			if (i == 0) {
				current = readFirstAnnotationBo(buffer);
				// save first annotation for delta bitfield
				// decodingContext.setPrevFirstAnnotationBo(current);
			} else {
				current = readDeltaAnnotationBo(buffer, prev);
			}

			prev = current;
			annotationBoList.add(current);
		}
		return annotationBoList;
	}

	private AnnotationBean readFirstAnnotationBo(Buffer buffer) {
		AnnotationBean current;
		current = new AnnotationBean();
		current.setKey(buffer.readSVInt());

		byte valueType = buffer.readByte();
		byte[] valueBytes = buffer.readPrefixedBytes();
		Object value = transcoder.decode(valueType, valueBytes);

		current.setValue(value);
		return current;
	}

	private List<SpanEventBean> readSpanEvent(Buffer buffer, SpanEventBean firstSpanEvent,
			SpanDecodingContext decodingContext) {
		final int spanEventSize = buffer.readVInt();
		if (spanEventSize <= 0) {
			return new ArrayList<SpanEventBean>();
		}
		final List<SpanEventBean> spanEventBoList = new ArrayList<SpanEventBean>();
		SpanEventBean prev = null;
		for (int i = 0; i < spanEventSize; i++) {
			SpanEventBean spanEvent;
			if (i == 0) {
				spanEvent = readFirstSpanEvent(buffer, firstSpanEvent, decodingContext);
			} else {
				spanEvent = readNextSpanEvent(buffer, prev, decodingContext);
			}
			prev = spanEvent;
			spanEventBoList.add(spanEvent);
		}

		return spanEventBoList;
	}

	private SpanEventBean readFirstSpanEvent(Buffer buffer, SpanEventBean firstSpanEvent,
			SpanDecodingContext decodingContext) {
		SpanEventBitField bitField = new SpanEventBitField(buffer.readByte());

		firstSpanEvent.setStartElapsed(buffer.readVInt());
		firstSpanEvent.setEndElapsed(buffer.readVInt());

		firstSpanEvent.setSequence(buffer.readShort());
		firstSpanEvent.setDepth(buffer.readSVInt());
		firstSpanEvent.setServiceType(buffer.readShort());

		if (bitField.isSetRpc()) {
			firstSpanEvent.setRpc(buffer.readPrefixedString());
		}

		if (bitField.isSetEndPoint()) {
			firstSpanEvent.setEndPoint(buffer.readPrefixedString());
		}
		if (bitField.isSetDestinationId()) {
			firstSpanEvent.setDestinationId(buffer.readPrefixedString());
		}

		firstSpanEvent.setApiId(buffer.readSVInt());

		if (bitField.isSetNextSpanId()) {
			firstSpanEvent.setNextSpanId(buffer.readLong());
		}

		if (bitField.isSetHasException()) {
			int exceptionId = buffer.readSVInt();
			String exceptionMessage = buffer.readPrefixedString();
			firstSpanEvent.setExceptionInfo(exceptionId, exceptionMessage);
		}

		if (bitField.isSetAnnotation()) {
			List<AnnotationBean> annotationBoList = readAnnotationList(buffer, decodingContext);
			firstSpanEvent.setAnnotationBoList(annotationBoList);
		}

		if (bitField.isSetNextAsyncId()) {
			firstSpanEvent.setNextAsyncId(buffer.readSVInt());
		}

		// if (bitField.isSetAsyncId()) {
		// firstSpanEvent.setAsyncId(buffer.readInt());
		// firstSpanEvent.setAsyncSequence((short) buffer.readVInt());
		// }
		return firstSpanEvent;
	}

	private SpanEventBean readNextSpanEvent(final Buffer buffer, final SpanEventBean prev,
			SpanDecodingContext decodingContext) {
		final SpanEventBean spanEventBo = new SpanEventBean();

		final SpanEventBitField bitField = new SpanEventBitField(buffer.readShort());

		switch (bitField.getStartElapsedEncodingStrategy()) {
		case PREV_DELTA:
			int startTimeDelta = buffer.readVInt();
			int startTime = startTimeDelta + prev.getStartElapsed();
			spanEventBo.setStartElapsed(startTime);
			break;
		case PREV_EQUALS:
			spanEventBo.setStartElapsed(prev.getStartElapsed());
			break;
		default:
			throw new IllegalStateException("unsupported SequenceEncodingStrategy");
		}
		spanEventBo.setEndElapsed(buffer.readVInt());

		switch (bitField.getSequenceEncodingStrategy()) {
		case PREV_DELTA:
			int sequenceDelta = buffer.readVInt();
			final int sequence = sequenceDelta + prev.getSequence();
			spanEventBo.setSequence((short) sequence);
			break;
		case PREV_ADD1:
			spanEventBo.setSequence((short) (prev.getSequence() + 1));
			break;
		default:
			throw new IllegalStateException("unsupported SequenceEncodingStrategy");
		}

		switch (bitField.getDepthEncodingStrategy()) {
		case RAW:
			spanEventBo.setDepth(buffer.readSVInt());
			break;
		case PREV_EQUALS:
			spanEventBo.setDepth(prev.getDepth());
			break;
		default:
			throw new IllegalStateException("unsupported DepthEncodingStrategy");
		}

		switch (bitField.getServiceTypeEncodingStrategy()) {
		case RAW:
			spanEventBo.setServiceType(buffer.readShort());
			break;
		case PREV_EQUALS:
			spanEventBo.setServiceType(prev.getServiceType());
			break;
		default:
			throw new IllegalStateException("unsupported ServiceTypeEncodingStrategy");
		}

		spanEventBo.setApiId(buffer.readSVInt());

		if (bitField.isSetRpc()) {
			spanEventBo.setRpc(buffer.readPrefixedString());
		}

		if (bitField.isSetEndPoint()) {
			spanEventBo.setEndPoint(buffer.readPrefixedString());
		}
		if (bitField.isSetDestinationId()) {
			spanEventBo.setDestinationId(buffer.readPrefixedString());
		}

		if (bitField.isSetNextSpanId()) {
			spanEventBo.setNextSpanId(buffer.readLong());
		}

		if (bitField.isSetHasException()) {
			int exceptionId = buffer.readSVInt();
			String exceptionMessage = buffer.readPrefixedString();
			spanEventBo.setExceptionInfo(exceptionId, exceptionMessage);
		}

		if (bitField.isSetAnnotation()) {
			List<AnnotationBean> annotationBoList = readAnnotationList(buffer, decodingContext);
			spanEventBo.setAnnotationBoList(annotationBoList);
		}

		if (bitField.isSetNextAsyncId()) {
			spanEventBo.setNextAsyncId(buffer.readSVInt());
		}

		if (bitField.isSetAsyncId()) {
			spanEventBo.setAsyncId(buffer.readInt());
			spanEventBo.setAsyncSequence((short) buffer.readVInt());
		}

		return spanEventBo;
	}

	private AnnotationBean readDeltaAnnotationBo(Buffer buffer, AnnotationBean prev) {
		AnnotationBean annotation = new AnnotationBean();

		final int prevKey = prev.getKey();

		annotation.setKey(buffer.readSVInt() + prevKey);

		byte valueType = buffer.readByte();
		byte[] valueBytes = buffer.readPrefixedBytes();
		Object value = transcoder.decode(valueType, valueBytes);

		annotation.setValue(value);
		return annotation;
	}

	private SpanChunkBean readSpanChunk(Buffer qualifier, Buffer columnValue, SpanDecodingContext decodingContext) {
		final SpanChunkBean spanChunk = new SpanChunkBean();

		final TransactionId transactionId = decodingContext.getTransactionId();
		spanChunk.setTransactionId(transactionId);
		spanChunk.setCollectorAcceptTime(decodingContext.getCollectorAcceptedTime());

		SpanEventBean firstSpanEvent = readQualifier(spanChunk, qualifier);
		readSpanChunkValue(columnValue, spanChunk, firstSpanEvent, decodingContext);

		return spanChunk;
	}

	private void readSpanChunkValue(Buffer buffer, SpanChunkBean spanChunk, SpanEventBean firstSpanEvent,
			SpanDecodingContext decodingContext) {
		final byte version = buffer.readByte();
		if (version != 0) {
			throw new IllegalStateException("unknown version :" + version);
		}
		spanChunk.setVersion(version);

		List<SpanEventBean> spanEventBoList = readSpanEvent(buffer, firstSpanEvent, decodingContext);
		spanChunk.addSpanEventBoList(spanEventBoList);
	}
	  public void next(SpanDecodingContext decodingContext) {
	        decodingContext.next();
	    }


}
